package com.shujia.flink.core

import org.apache.flink.streaming.api.scala._

object Demo1StreamWordCount {
  def main(args: Array[String]): Unit = {
    /**
      * 创建flink环境
      *
      */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * 读取socket 构建DS
      *
      * nc -lk 8888
      */
    val lineDS: DataStream[String] = env.socketTextStream("master",8888)

    //1、将单词拆分
    val wordDS: DataStream[String] = lineDS.flatMap(_.split(","))


    //2、转化成kv格式
    val kvDS: DataStream[(String, Int)] = wordDS.map((_,1))

    //3、按照key进行分组,底层也是hash分区
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    //4、对value进行聚合,通过下标进行聚合
    val countDS: DataStream[(String, Int)] = keyByDS.sum(1)

    //打印这个结果

    //启动flink程序
    env.execute()
  }
}
